import com.atguigu.gmall.realtime.app.BaseApp;
import org.apache.doris.flink.cfg.DorisExecutionOptions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisReadOptions;
import org.apache.doris.flink.sink.DorisSink;
import org.apache.doris.flink.sink.writer.SimpleStringSerializer;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Properties;

/**
 * @Author lzc
 * @Date 2023/7/11 13:46
 */
public class DorisStream extends BaseApp {
    public static void main(String[] args) {
        new DorisStream().start(
            3000,
            2,
            "DorisStream",
            "ods_db"
        );
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env,
                       DataStreamSource<String> stream) {
    
        /*DorisOptions.Builder builder = DorisOptions.builder()
            .setFenodes("hadoop162:7030")
            .setTableIdentifier("test.table1")
            .setUsername("root")
            .setPassword("aaaaaa");
    
        DorisSource<List<?>> dorisSource = DorisSourceBuilder.<List<?>>builder()
            .setDorisOptions(builder.build())
            .setDorisReadOptions(DorisReadOptions.builder().build())
            .setDeserializer(new SimpleListDeserializationSchema())
            .build();
    
        DataStreamSource<List<?>> source = env.fromSource(dorisSource, WatermarkStrategy.noWatermarks(), "doris-source");
        source.print();
    */
    
        Properties props = new Properties();
        props.setProperty("format", "json");
        props.setProperty("read_json_by_line", "true"); // 每行一条 json 数据
        DorisSink<String> sink = DorisSink.<String>builder()
            .setDorisReadOptions(DorisReadOptions.builder().build())
            .setDorisOptions(DorisOptions.builder()
                                 .setFenodes("hadoop162:7030")
                                 .setTableIdentifier("test.table1")
                                 .setUsername("root")
                                 .setPassword("aaaaaa")
                                 .build()
            )
            .setDorisExecutionOptions(DorisExecutionOptions.builder()
                                          .setLabelPrefix("a")
                                          //                                          .disable2PC()  // 禁用两阶段提交: 如果不禁用LabelPrefix必须全局唯一
                                          .setBufferCount(3)
                                          .setBufferSize(1024 * 1024)
                                          .setCheckInterval(3000)
                                          .setMaxRetries(3)
                                          .setStreamLoadProp(props)
                                          .build())
            .setSerializer(new SimpleStringSerializer())
            .build();
        
        stream.sinkTo(sink);
        
        
    }
}
